package com.flink.timewindow.window;

import com.flink.timewindow.bean.WaterSensor;
import com.flink.timewindow.function.WaterSensorMapFunction;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SessionWindowTimeGapExtractor;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


public class TimeWindowDemo {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //获取数据源
        SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("10.90.100.102", 8888)
                //数据处理
                //切分转换
                .map(new WaterSensorMapFunction());
        //分组
        KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(value -> value.getId());

        //窗口分配器
        WindowedStream<WaterSensor, String, TimeWindow> sensorWS =
                //滚动窗口  窗口长度10s
                sensorKS.window( TumblingProcessingTimeWindows.of(Time.seconds(10)) );
                //滑动窗口  长度10s  步长5s
               // sensorKS.window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5)));
                //会话窗口  间隔5s
               //sensorKS.window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)));
                //会话窗口，动态间隔，每一条来的数据都会更新间隔时间
//                sensorKS.window(ProcessingTimeSessionWindows.withDynamicGap(new SessionWindowTimeGapExtractor<WaterSensor>() {
//                    @Override
//                    public long extract(WaterSensor element) {
//                        //从数据中提取ts作为时间间隔，单位ms
//                        return element.getTs()*1000L;
//                    }
//                }));


        //窗口函数
        SingleOutputStreamOperator<String> process = sensorWS.process(
                // 泛型： 1：输入类型  2：输出类型  3：分组key类型  4：窗口类型
                new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {

                    //全局函数计算逻辑： 窗口出发时才会调用一次，统一计算窗口的所有数据
                    //s: 分组的key  context：上下文   elements：存的数据   out：采集器
                    @Override
                    public void process(String s,
                                        ProcessWindowFunction<WaterSensor, String, String, TimeWindow>.Context context,
                                        Iterable<WaterSensor> elements,
                                        Collector<String> out) throws Exception {
                        //上下文可以拿到window对象，还有其他东西：如测流输出
                        long count = elements.spliterator().estimateSize();
                        long windowStartTs = context.window().getStart();
                        long windowEndTs = context.window().getEnd();
                        String windowStart = DateFormatUtils.format(windowStartTs, "yyyy-MM-dd HH:mm:ss.SSS");
                        String windowEnd = DateFormatUtils.format(windowEndTs, "yyyy-MM-dd HH:mm:ss.SSS");

                        out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());



                    }
                }
        );
        //输出
        process.print();
        //执行
        env.execute();

    }
}

/***思考***
 * 触发器、移除器：现成的几个窗口，都有默认的实现，一般不需要自定义
 * 以时间类型的滚动窗口为例分析原理
  TODO 1. 窗口什么时候触发输出？
  时间进展>=窗口的最大时间戳（end - 1ms）

  TODO 2. 窗口是如何划分的？
 start= 向下取整，取窗口长度的整数倍
 end = start + 窗口长度
 窗口左闭右开 ： 属于本窗口的 最大时间戳 = end -1ms

 TODO 3.窗口生命周期？
 创建： 属于本窗口的第一条数据来的时候，现new的，放入一个singleton单例的集合中
 销毁（关窗）： 时间进展>=窗口的最大时间戳（end - 1ms）+ 允许迟到的时间（默认0）




 */

